package com.quanyan.operative.mq.Producer;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import com.quanyan.mq.client.reqeust.SendCallbackListener;
import com.quanyan.mq.client.response.MqResponse;
import com.quanyan.mq.enums.MqSendStatus;
import com.quanyan.operative.constants.OperativeConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * 生成夺宝号生产者
 * Created by huadongyang on 2017/1/3.
 */
@Component("generateSnatchNumProducer")
public class GenerateSnatchNumProducer {
    private static final Logger log = LoggerFactory.getLogger(GenerateSnatchNumProducer.class);

    @Value("${mq.nameServer.address}")
    private String nameServerAddress;

    private DefaultMQProducer producer;


    protected String getProducerGroup() {
        return OperativeConstant.GENERATE_SNATCH_NUM_GROUP;
    }

    protected String getNameServerAddress() {
        return this.nameServerAddress;
    }

    protected String getInstanceName() {
        return OperativeConstant.GENERATE_SNATCH_NUM_INSTANCE;
    }

    void initProducer() throws MQClientException {
        producer = new DefaultMQProducer(getProducerGroup());//保证应用内唯一
        producer.setNamesrvAddr(getNameServerAddress());
        producer.setInstanceName(getInstanceName());//保证应用内唯一
    }

    @PostConstruct
    void run() {
        try {
            log.info("夺宝消息生产者开始启动");
            initProducer();  //初始化生产者
            producer.start(); // 启动生产者
            log.info("夺宝消息生产者启动成功");
        } catch (MQClientException e) {
            log.error(e.getMessage());
        }

        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                producer.shutdown();
                log.info("producer调用addShutdownHook方法优雅关闭 ！！！");
            }
        }));
    }

    /**
     * 异步发送
     *
     * @param msg
     * @param sendCallbackListener
     */
    public void asyncSend(Message msg, final SendCallbackListener sendCallbackListener) {
        try {
            initProducer();
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info(JSON.toJSONString(sendResult.getSendStatus()));
                    if (null == sendCallbackListener) {
                        return;
                    }
                    sendCallbackListener.onSuccess(JSON.toJSONString(sendResult.getSendStatus()));
                }

                @Override
                public void onException(Throwable e) {
                    log.error(e.getMessage());
                    if (null == sendCallbackListener) {
                        return;
                    }
                    sendCallbackListener.onException(e);
                }
            });
            log.info("send message to " + msg.getTopic() + " key: " + msg.getKeys());

        } catch (Exception e) {
            sendCallbackListener.onException(e);
            e.printStackTrace();
            log.error("send message to " + msg.getTopic() + " key: " + msg.getKeys() +
                    "sendException : " + e.getMessage());
        }
    }


    /**
     * 同步发送
     *
     * @param msg
     * @return
     */
    public MqResponse syncSend(Message msg) {
        msg.getBody();

        SendResult sendResult = null;
        MqResponse mqResponse = new MqResponse();
        mqResponse.setMqSendStatus(MqSendStatus.FAIL);

        try {
            sendResult = producer.send(msg);//send方法支持内部重试
            if (sendResult == null) {
                return mqResponse;
            }
            SendStatus sendStatus = sendResult.getSendStatus();

            if (sendStatus.equals(SendStatus.SEND_OK)) {
                mqResponse.setMqSendStatus(MqSendStatus.SEND_OK);
            }

            log.info("send message to " + msg.getTopic() + " key: " + msg.getKeys() +
                    "sendResult : " + com.alibaba.fastjson.JSON.toJSONString(sendResult));

        } catch (Exception e) {

            //如果消息不可丢,必须重发,或者存储到Cache/Db中,由定时器调用重发.
            log.error("send message to " + msg.getTopic() + " key: " + msg.getKeys() +
                    "sendResult : " + com.alibaba.fastjson.JSON.toJSONString(sendResult) +
                    "sendException : " + e.getMessage());
        }

        return mqResponse;
    }
}
